

#include "ParallelMemoryManager.h"


ParallelMemoryManager::ParallelMemoryManager(MemoryTable *parMemoryTable, int parCollectorThreshold, int parNumClusters, float parLastThreshold){

	memoryTable = parMemoryTable;
	memoryTableToReset = false;
	collector = new Collector();
	encoder = new Encoder(parNumClusters);
	collectorThreshold = parCollectorThreshold;
	lastThreshold = parLastThreshold;
	inputEventCond = new pthread_cond_t;
	pthread_cond_init(inputEventCond, NULL);
	inputEventMutex = new pthread_mutex_t;
	pthread_mutex_init(inputEventMutex, NULL);
	inputSubscriptionCond = new pthread_cond_t;
	pthread_cond_init(inputSubscriptionCond, NULL);
	inputSubscriptionMutex = new pthread_mutex_t;
	pthread_mutex_init(inputSubscriptionMutex, NULL);
	encoderMutex = new pthread_mutex_t;
	pthread_mutex_init(encoderMutex, NULL);
	subscriptionMapMutex = new pthread_mutex_t;
	pthread_mutex_init(subscriptionMapMutex, NULL);
	eventMapMutex = new pthread_mutex_t;
	pthread_mutex_init(eventMapMutex, NULL);
	memoryTableMutex = new pthread_mutex_t;
	pthread_mutex_init(memoryTableMutex, NULL);
	eventThread = new pthread_t;
	subscriptionThread = new pthread_t;
	updatingThread = new pthread_t;
	pthread_create(eventThread, NULL, parallel_manager::startEventThread, (void *) this);
	pthread_create(subscriptionThread, NULL, parallel_manager::startSubscriptionThread, (void *) this);
	pthread_create(updatingThread, NULL, parallel_manager::startUpdatingThread, (void *) this);

}

ParallelMemoryManager::~ParallelMemoryManager(){

	pthread_cond_destroy(inputEventCond);
	pthread_mutex_destroy(inputEventMutex);
	pthread_cond_destroy(inputSubscriptionCond);
	pthread_mutex_destroy(inputSubscriptionMutex);
	delete inputEventCond;
	delete inputEventMutex;
	delete inputSubscriptionCond;
	delete inputSubscriptionMutex;
	while (! inputEventQueue.empty()) {
			PubPkt *pkt = inputEventQueue.front();
			inputEventQueue.pop();
			if (pkt->decRefCount()) delete pkt;
	}
	while (! inputSubscriptionQueue.empty()) {
			BackSubPkt *pkt = inputSubscriptionQueue.front();
			inputSubscriptionQueue.pop();
		/*	if (pkt->decRefCount()) */ delete pkt;
	}
	delete collector;
	delete encoder;
	pthread_mutex_destroy(encoderMutex);
	pthread_mutex_destroy(subscriptionMapMutex);
	pthread_mutex_destroy(eventMapMutex);
	pthread_mutex_destroy(memoryTableMutex);
	delete encoderMutex;
	delete subscriptionMapMutex;
	delete eventMapMutex;
	delete memoryTableMutex;
	pthread_cancel(*eventThread);
	delete eventThread;
	pthread_cancel(*subscriptionThread);
	delete subscriptionThread;
	pthread_cancel(*updatingThread);
	delete updatingThread;
}


void ParallelMemoryManager::addEvent(PubPkt *pkt){

	pthread_mutex_lock(inputEventMutex);
	inputEventQueue.push(pkt);
	pthread_cond_signal(inputEventCond);
	pthread_mutex_unlock(inputEventMutex);
}

void ParallelMemoryManager::addSubscription(BackSubPkt *pkt){

	pthread_mutex_lock(inputSubscriptionMutex);
	inputSubscriptionQueue.push(pkt);
	pthread_cond_signal(inputSubscriptionCond);
	pthread_mutex_unlock(inputSubscriptionMutex);
}


void ParallelMemoryManager::queryEventSet(BackSubPkt *pkt, Outbox &outbox){

	set<PubPkt*> resultSet;
	pthread_mutex_lock(memoryTableMutex);
	memoryTable->computeExtraction(pkt, outbox.pkts);
	pthread_mutex_unlock(memoryTableMutex);
//	return resultSet;
}


void ParallelMemoryManager::manageSubscription(){

	while (true) {
		pthread_mutex_lock(inputSubscriptionMutex);
		if (inputSubscriptionQueue.empty()) {
			pthread_cond_wait(inputSubscriptionCond, inputSubscriptionMutex);
		}
		BackSubPkt *pkt = inputSubscriptionQueue.front();
		inputSubscriptionQueue.pop();
		pthread_mutex_unlock(inputSubscriptionMutex);

		collector->inspectSubscription(pkt);
		pthread_mutex_lock(encoderMutex);
		set<string> clustersMatched = encoder->encode(pkt);
		pthread_mutex_unlock(encoderMutex);
		pthread_mutex_lock(subscriptionMapMutex);
		for (set<string>::iterator it=clustersMatched.begin(); it!=clustersMatched.end(); it++) {
			string complexType = *it;
			boost::unordered_map<string, SubscribtionInfo>::iterator it2 = subscriptionMap.find(complexType);
			if (it2==subscriptionMap.end()) {
				SubscribtionInfo subInfo;
				subInfo.windowsList.push_back(pkt->getWindow());
				if (pkt->getOccurrence() == LAST){
					subInfo.countLast = 1;
				} else {
					subInfo.countLast = 0;
				}
				subscriptionMap.insert(make_pair(complexType, subInfo));
			} else {
				it2->second.windowsList.push_back(pkt->getWindow());
				if (pkt->getOccurrence() == LAST){
					it2->second.countLast++;
				}
			}
		}
		clustersMatched.clear();
		pthread_mutex_unlock(subscriptionMapMutex);
		if (collector->getCount() == collectorThreshold){
			pthread_mutex_lock(encoderMutex);
			pthread_mutex_lock(eventMapMutex);
			pthread_mutex_lock(subscriptionMapMutex);
			encoder->clear();
			boost::unordered_map<int, boost::unordered_map<string, vector<Type> > > collectorMap = collector->getMap();
			for (boost::unordered_map<int, boost::unordered_map<string, vector<Type> > >::iterator it=collectorMap.begin(); it!=collectorMap.end(); it++) {
				encoder->addEntry(it->first, it->second);
			}
			subscriptionMap.clear();
			eventMap.clear();
			pthread_mutex_lock(memoryTableToResetMutex);
			memoryTableToReset = true;
			pthread_mutex_unlock(memoryTableToResetMutex);
			pthread_mutex_unlock(subscriptionMapMutex);
			pthread_mutex_unlock(eventMapMutex);
			pthread_mutex_unlock(encoderMutex);
		}
	}
}

void ParallelMemoryManager::manageEvents(){

	while (true) {
		pthread_mutex_lock(inputEventMutex);
		if (inputEventQueue.empty()) {
			pthread_cond_wait(inputEventCond, inputEventMutex);
		}
		PubPkt *pkt = inputEventQueue.front();
		inputEventQueue.pop();
		pthread_mutex_unlock(inputEventMutex);

		pthread_mutex_lock(encoderMutex);
		string complexType = encoder->encode(pkt);
		pthread_mutex_unlock(encoderMutex);
		pthread_mutex_lock(eventMapMutex);
		boost::unordered_map<string, int>::iterator it = eventMap.find(complexType);
		if (it==eventMap.end()) {
			eventMap.insert(make_pair(complexType, 1));
		} else {
			it->second++;
		}
		pthread_mutex_unlock(eventMapMutex);

		//* ORA DEVO ORDINARE L'INSERIMENTO ALLA TABLE *//
		pthread_mutex_lock(memoryTableMutex);
		memoryTable->computeInsertion(pkt, complexType);
		pthread_mutex_unlock(memoryTableMutex);
	}
}

void ParallelMemoryManager::updateTable(){

	while (true) {
		pthread_mutex_lock(eventMapMutex);
		boost::unordered_map<string, int> tempEventMap = eventMap;
		pthread_mutex_unlock(eventMapMutex);
		pthread_mutex_lock(subscriptionMapMutex);
		boost::unordered_map<string, SubscribtionInfoStruct> tempSubscriptionMap = subscriptionMap;
		pthread_mutex_unlock(subscriptionMapMutex);
		pthread_mutex_lock(memoryTableMutex);
		pthread_mutex_lock(memoryTableToResetMutex);
		if (memoryTableToReset){
			memoryTable->resetTable();
		}
		pthread_mutex_unlock(memoryTableToResetMutex);
		for(boost::unordered_map<string, SubscribtionInfo>::iterator it = tempSubscriptionMap.begin(); it!=tempSubscriptionMap.end(); it++){
			string complexType = it->first;
			int eventsCount = tempEventMap[complexType];
			vector<float> windowsVector = it->second.windowsList;
			int subsCount = windowsVector.size();
			sort(windowsVector.begin(), windowsVector.end());
			float median = windowsVector[(int)windowsVector.size()/2];
			bool resettable;
			float percentage = it->second.countLast / subsCount;
			if (percentage < lastThreshold){
				resettable = true;
			} else {
				resettable = false;
			}
			memoryTable->updateTable(complexType, subsCount, median, resettable, eventsCount);
		}
		pthread_mutex_unlock(memoryTableMutex);
	}
}


void * parallel_manager::startEventThread(void *input){
	ParallelMemoryManager *manager = (ParallelMemoryManager*) input;
	manager->manageEvents();
	pthread_exit(NULL);
	return 0;
}


void * parallel_manager::startSubscriptionThread(void *input){
	ParallelMemoryManager *manager = (ParallelMemoryManager*) input;
	manager->manageSubscription();
	pthread_exit(NULL);
	return 0;
}


void * parallel_manager::startUpdatingThread(void *input){
	ParallelMemoryManager *manager = (ParallelMemoryManager*) input;
	manager->updateTable();
	pthread_exit(NULL);
	return 0;
}







